package com.bridgeintelligent.tag.bulkload.service.looklike;

import com.bridgeintelligent.tag.utils.DateHelper;
import com.wayneleo.quickstart.QuickException;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.FileSystem;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.List;

import static com.bridgeintelligent.tag.bulkload.api.exception.BulkLoadErrors.ERROR_401008;
import static com.bridgeintelligent.tag.constants.PublicConstant.TXT;

@Service
@Slf4j
public class HdfsService {


    @Value( "${hdfs.upload.path:/data/hdfs/upload}" )
    String hdfsUploadPath;

    @Value( "${hdfs.upload.static.path:/user/cpmuser/staticustomers}" )
    String hdfsUploadStaticPath;

    @Value( "${hdfs.upload.dimfree.path:/user/cpmuser/dimfree}" )
    String hdfsUploadDimfreePath;

    public void saveDataToHdfs( String fileName, List<String> data ) throws Exception{
        FileSystem fileSystem = HdfsUtil.getFileSystem();
        String pathFile =  hdfsUploadPath + "/" + fileName.substring(0,fileName.lastIndexOf("."));
        HdfsUtil.mkdir(fileSystem,pathFile);
        fileName = pathFile+ "/" + fileName;
        HdfsUtil.writeDate(fileSystem,data,fileName);
    }

    public String getPath(String fileName,String esIndex) throws Exception {
        FileSystem fileSystem = HdfsUtil.getFileSystem();
        String pathFile =  hdfsUploadPath + "/" + esIndex;
        HdfsUtil.mkdir(fileSystem,pathFile);
        fileName = pathFile+ "/" + fileName;
        return fileName;
    }

    public String getSaticPath(String fileName) throws Exception {
        FileSystem fileSystem = HdfsUtil.getFileSystem();
        String pathFile =  hdfsUploadStaticPath;
        HdfsUtil.mkdir(fileSystem,pathFile);
        fileName = pathFile+ "/" + fileName;
        return fileName;
    }
    public String getHdfsPath(String fileName,String path) throws Exception {
        FileSystem fileSystem = HdfsUtil.getFileSystem();
        String pathFile =  path;
        HdfsUtil.mkdir(fileSystem,pathFile);
        fileName = pathFile+ "/" + fileName;
        return fileName;
    }

    /**
     * 本地文件上传到Hdfs的公共方法
     * @param localFilePath 本地文件全路径
     * @param type 上传到hdfs文件夹的类型 不同的文件类型对应不同的文件夹目录 开发人员自己配置
     * @return
     * @throws Exception
     */
    public String uploadHdfs(String localFilePath, int type) throws Exception {
        String hdfsPath;
        String fileName = localFilePath.substring(localFilePath.lastIndexOf(File.separator)+1);
        switch (type){
            case 1:
                String id = fileName.substring(0,fileName.lastIndexOf("."));
                String dimFreeFileName = id+"_"+ DateHelper.currentDateTime(DateHelper.PATTERN_DATE_8)+TXT;
                hdfsPath = this.getHdfsPath(dimFreeFileName,hdfsUploadDimfreePath);
                break;
            default:
                hdfsPath = this.getHdfsPath(fileName,hdfsUploadPath);
        }
        log.info("=====>本地备份文件【{}】开始写入HDFS",localFilePath);
        this.upload(localFilePath,hdfsPath);
        log.info("=====>备份文件向HDFS写入的全路径：{}",hdfsPath);
        return hdfsPath;
    }

    private void upload(String localPath,String remotePath){
        long startTime = System.currentTimeMillis();
        try {
            HdfsUtil.uploadFile(HdfsUtil.getFileSystem(),localPath,remotePath);
        } catch (Exception e) {
            throw QuickException.normal(ERROR_401008.args(remotePath),e);
        }
        long endTime = System.currentTimeMillis();
        log.info("=====>备份文件上传hdfs[{}]成功，耗时：{}s",localPath.substring(localPath.lastIndexOf(File.separator)+1),(endTime - startTime)/1000);
    }
}
